package cn.wangjie.spark.process

import java.util.Properties

import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}
import org.apache.spark.storage.StorageLevel

/**
 * 对电影评分数据进行统计分析，获取Top10电影（电影评分平均值最高，并且每个电影被评分的次数大于2000)。
 */
object SparkTop10Movie {
	
	def main(args: Array[String]): Unit = {
		
		// 构建SparkSession实例对象
		val spark: SparkSession = SparkSession.builder()
    		.appName(this.getClass.getSimpleName.stripSuffix("$"))
			// TODO: 设置SparkSQL产生Shuffle时分区数目
    		.config("spark.sql.shuffle.partitions", "4")
    		.master("local[2]")
    		.getOrCreate()
		import spark.implicits._
		
		// TODO: 第一步、读取电影评分数据，从本地文件系统读取
		val rawRatingsDS: Dataset[String] = spark.read.textFile("datas/ml-1m/ratings.dat")
		
		
		// TODO: 第二步、转换数据，指定Schema信息，封装到DataFrame
		val ratingsDF: DataFrame = rawRatingsDS
			// 过滤不合格的数据
			.filter(line => null != line && line.trim.split("::").length == 4 )
			// 数据转换
			.mapPartitions{iter =>
				iter.map{line =>
					// 按照分割符分割，拆箱到变量中
					val Array(userId, movieId, rating, timestamp) = line.trim.split("::")
					// 返回四元组
					(userId, movieId, rating.toDouble, timestamp.toLong)
				}
			}
			// 指定列名称，转换为DataFrame
			.toDF("userId", "movieId", "rating", "timestamp")
		//ratingsDF.printSchema()
		//ratingsDF.show(10, truncate = false)
		
		// TODO: 第三步、基于SQL方式分析
		// step1. 将DataFrame注册为临时视图
		ratingsDF.createOrReplaceTempView("view_tmp_ratings")
		// step2. 编写SQL，执行分析
		val resultDF: DataFrame = spark.sql(
			"""
			  |SELECT
			  |  movieId, COUNT(1) AS cnt, ROUND(AVG(rating), 2) AS avg_rating
			  |FROM
			  |  view_tmp_ratings
			  |GROUP BY
			  |  movieId
			  |HAVING
			  |  cnt > 2000
			  |ORDER BY
			  |  avg_rating DESC, cnt DESC
			  |LIMIT
			  |  10
			  |""".stripMargin)
		//resultDF.printSchema()
		//resultDF.show(10, truncate = false)
		
		//println("============================================================")
		
		// TODO: 第四步、基于DSL方式分析
		import org.apache.spark.sql.functions._
		val top10MovieDF: Dataset[Row] = ratingsDF
			// 按照movieId分组
			.groupBy($"movieId")
			// 聚合操作，统计个数和平均值
			.agg( //
				count($"movieId").as("cnt"),  //
				round(avg($"rating"), 2).as("avg_rating") //
			)
			// 过滤评分次数大于2000
			.where($"cnt".gt(2000))
			// 降序排序
			.orderBy(
				$"avg_rating".desc, $"cnt".desc //
			)
			// 获取前10个
			.limit(10)
		//top10MovieDF.printSchema()
		//top10MovieDF.show(10, truncate = false)
		
		// TODO: 将结果数据保存至MySQL表中和CSV文件中
		
		// 第一、将结果DataFrame进行缓存
		top10MovieDF.persist(StorageLevel.MEMORY_AND_DISK)
		
		// 第二、保存至MySQL表中
		top10MovieDF
            .coalesce(1) // 降低分区数目
			.write
            .mode(SaveMode.Overwrite)
            .option("driver", "com.mysql.cj.jdbc.Driver")
            .option("user", "root")
            .option("password", "123456")
            .jdbc(
			   "jdbc:mysql://node1.itcast.cn:3306/?serverTimezone=UTC&characterEncoding=utf8&useUnicode=true", //
			    "db_test.tb_top10_movies",
			    new Properties()
		    )
		
		// 第三、保存CSV文件
		top10MovieDF
			.coalesce(1) // 降低分区数目
			.write
			.mode(SaveMode.Overwrite)
			.csv("datas/spark/top10movies/")
		
		// 释放资源
		top10MovieDF.unpersist()
		
		Thread.sleep(10000000)
		
		// 应用结束关闭资源
		spark.stop()
	}
	
}
